package com.mxnavi5.example.DataStream;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.co.CoProcessFunction;
import org.apache.flink.streaming.api.functions.co.KeyedBroadcastProcessFunction;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.ProcessAllWindowFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

import java.time.Duration;
import java.util.Iterator;
import java.util.concurrent.TimeUnit;

public class MyProcessTest {
    private static final OutputTag<Integer> outputTag = new OutputTag<Integer>("OutputTag") {
    };

    public static void main(String[] args) throws Exception {
        //    StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        Configuration configuration = new Configuration();
        configuration.setInteger(RestOptions.PORT, 8081);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);
        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

        DataStreamSource<Integer> integerDataStreamSource = env.fromElements(1, 2, 3);


//        SingleOutputStreamOperator<Integer> mapStream = integerDataStreamSource.map(x -> x * 2);
//        mapStream.print("map");


//        SingleOutputStreamOperator<Integer> flatMapStream = processStream.flatMap(new FlatMapFunction<Integer, Integer>() {
//            @Override
//            public void flatMap(Integer integer, Collector<Integer> collector) throws Exception {
//                collector.collect(integer + 1);
//                collector.collect(integer + 1);
//            }
//        });

//        flatMapStream.print("flatMap");

        //process   自定义逻辑
        SingleOutputStreamOperator<Integer> processStream = integerDataStreamSource.process(new ProcessFunction<Integer, Integer>() {
            @Override
            public void processElement(Integer value, ProcessFunction<Integer, Integer>.Context ctx, Collector<Integer> out) throws Exception {
                if (value < 3) {
                    out.collect(value * 2 + 1);
                    out.collect(value * 2 + 1);
                } else {
                    //侧输出流
                    ctx.output(outputTag, value * 2 + 1);
                }
            }
        });
//        processStream.print("prcess");
//        processStream.getSideOutput(outputTag).print("outputTag");


        SingleOutputStreamOperator<Integer> dataStreamSource = env.addSource(new UdfSourceTest()).returns(Integer.class);
        SingleOutputStreamOperator<Tuple2<Integer, Integer>> mapStream = dataStreamSource.map(x -> Tuple2.of(1, x))
                .returns(TypeInformation.of(new TypeHint<Tuple2<Integer, Integer>>() {
                }));

        KeyedStream<Tuple2<Integer, Integer>, Integer> keyStream = mapStream.keyBy(t -> t.f0);


        //KeyedProcessFunction
        keyStream.process(new KeyedProcessFunction<Integer, Tuple2<Integer, Integer>, String>() {

            @Override
            public void processElement(Tuple2<Integer, Integer> value, KeyedProcessFunction<Integer, Tuple2<Integer, Integer>, String>.Context ctx, Collector<String> out) throws Exception {
                out.collect("key:"+ctx.getCurrentKey()+",value"+value);
                //注册定时器
                if(value.f1==5) {
                    long l = ctx.timerService().currentProcessingTime();
                    ctx.timerService().registerProcessingTimeTimer(l+5 * 1000);
                }
            }


            @Override
            public void onTimer(long timestamp, KeyedProcessFunction<Integer, Tuple2<Integer, Integer>, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
                System.out.println("@@@@@@@@@@@@@@@@@@@@@@@@@@@@@");
            }
        }).print("keyedProcess");

        //ProcessWindowFunction
        SingleOutputStreamOperator tumblingSum = keyStream
                .window(TumblingProcessingTimeWindows.of(Time.of(5, TimeUnit.SECONDS)))
                .process(new ProcessWindowFunction<Tuple2<Integer, Integer>, String, Integer, TimeWindow>() {
                    @Override
                    public void process(Integer integer, ProcessWindowFunction<Tuple2<Integer, Integer>, String, Integer, TimeWindow>.Context context, Iterable<Tuple2<Integer, Integer>> elements, Collector<String> out) throws Exception {
                        String s ="elements:";
                        Iterator<Tuple2<Integer, Integer>> iterator = elements.iterator();
                        while (iterator.hasNext()){
                            s=s+iterator.next().f1+",";
                        }
                        s = s+"start:"+context.window().getStart()+",end:"+context.window().getEnd();
                        out.collect(s);
                    }

                });
//        tumblingSum.print("TumblingSum:");








        //关联
        DataStreamSource<Tuple2<Integer, Integer>> tuple2_1DataStreamSource
                = env.fromElements(Tuple2.of(1, 1), Tuple2.of(1, 2),
                Tuple2.of(2, 1), Tuple2.of(2, 2));

//        tuple2_1DataStreamSource
//                .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple2<Integer, Integer>>(Time.milliseconds(0)) {
//                    @Override
//                    public long extractTimestamp(Tuple2<Integer, Integer> element) {
//                        return element.f1;
//                    }
//                });



        KeyedStream<Tuple2<Integer, Integer>, Integer> keyed1Stream = tuple2_1DataStreamSource
                .keyBy(x -> x.f0);

        DataStreamSource<Tuple2<Integer, Integer>> tuple2_2DataStreamSource
                = env.fromElements(Tuple2.of(1, 3), Tuple2.of(1, 4),
                Tuple2.of(2, 5));

        KeyedStream<Tuple2<Integer, Integer>, Integer> keyed2Stream = tuple2_2DataStreamSource
                .keyBy(x -> x.f0);


//        SingleOutputStreamOperator<Tuple3<Integer, Integer, Integer>> coMapStream = tuple2_1DataStreamSource.connect(tuple2_2DataStreamSource).map(new CoMapFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple3<Integer, Integer, Integer>>() {
//            @Override
//            public Tuple3<Integer, Integer, Integer> map1(Tuple2<Integer, Integer> t2) throws Exception {
//                return Tuple3.of(t2.f0, t2.f1, 10);
//            }
//
//            @Override
//            public Tuple3<Integer, Integer, Integer> map2(Tuple2<Integer, Integer> t2) throws Exception {
//                return Tuple3.of(t2.f0, t2.f1, 20);
//            }
//        });
//        coMapStream.print("coMap");

        //CoProcessFunction
//        tuple2_1DataStreamSource.connect(tuple2_2DataStreamSource)
//                .process(new CoProcessFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, String>() {
//                    @Override
//                    public void processElement1(Tuple2<Integer, Integer> value, CoProcessFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, String>.Context ctx, Collector<String> out) throws Exception {
//                        out.collect("fromStream1,value:" + value);
//                    }
//
//                    @Override
//                    public void processElement2(Tuple2<Integer, Integer> value, CoProcessFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, String>.Context ctx, Collector<String> out) throws Exception {
//                        out.collect("fromStream2,value:" + value);
//                    }
//                }).print("CoProcess");


        //ProcessJoinFunction
//       keyed1Stream
//               .intervalJoin(keyed2Stream)
//               .between(Time.milliseconds(0), Time.milliseconds(1000))
//                       .process(new ProcessJoinFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, String>() {
//                           @Override
//                           public void processElement(Tuple2<Integer, Integer> left, Tuple2<Integer, Integer> right, ProcessJoinFunction<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, String>.Context ctx, Collector<String> out) throws Exception {
//                               out.collect("left:"+left+",right:"+right);
//                           }
//                       }).print("ProcessJoinFunction");







        env.execute();


    }

}
